package per.mapt.basic.nio;

import java.io.Closeable;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Arrays;
import java.util.Iterator;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;


public class NioClient implements Closeable {
    private final String addr;
    private final int port;
    private Selector selector;
    private SocketChannel channel;
    private DataWriter dataWriter;
    private boolean isConnect;
    private BlockingQueue<byte[]> reading = new ArrayBlockingQueue<>(1024);

    public NioClient(String addr, int port) {
        this.addr = addr;
        this.port = port;
        isConnect = false;
    }

    public void conect() throws IOException {
        channel = SocketChannel.open();
        channel.configureBlocking(false);
        channel.connect(new InetSocketAddress(addr, port));
        selector = Selector.open();
        channel.register(selector, SelectionKey.OP_CONNECT | SelectionKey.OP_READ);
        // 可以考虑多个共用一个writer
        dataWriter = new DataWriter("client");
        dataWriter.start();
        // TODO 使用新线程来处理
        new Thread(() -> {
            try {
                while(true) {
                    selector.select();
                    for (Iterator<SelectionKey> key = selector.selectedKeys().iterator(); key.hasNext(); ) {
                        SelectionKey selectionKey = key.next();
                        key.remove();
                        if (selectionKey.isConnectable()) {
                            SocketChannel keyChannel = (SocketChannel) selectionKey.channel();
                            keyChannel.finishConnect();
                            isConnect = true;
                        }
                        if (selectionKey.isReadable()) {
                            //  read data
                            SocketChannel keyChannel = (SocketChannel) selectionKey.channel();
                            ByteBuffer allocate = ByteBuffer.allocate(1024);
                            int readLength;
                            do {
                                allocate.clear();
                                readLength = keyChannel.read(allocate);
                                if (readLength > 0) {
                                    allocate.flip();
                                    byte[] dst = new byte[readLength];
                                    allocate.get(dst, 0, readLength);
                                    reading.offer(dst);
                                }
                            } while (readLength > 0);
                        }
                    }
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }).start();
    }

    public void writeData(byte[] content) {
        while (!channel.isConnected()) {
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        byte[] array = Arrays.copyOf(content, content.length);
        dataWriter.writeData(channel, ByteBuffer.wrap(array, array.length, 0));
    }

    public byte[] readData() {
        try {
            return reading.take();
        } catch (InterruptedException e) {
            return null;
        }
    }

    public byte[] tryReadData() {
        return reading.poll();
    }

    @Override
    public void close() throws IOException {
        selector.close();
        dataWriter.close();
        channel.close();
    }
}
